/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.partitioner.implementation;

import edu.isi.pegasus.planner.catalog.site.classes.SiteCatalogEntry;
import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PegasusBag;
import edu.isi.pegasus.planner.subworkflow.partitioner.DAXAbstraction;
import edu.isi.pegasus.planner.subworkflow.partitioner.PurePartitioner;
import java.util.Iterator;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

/**
 *
 * @author wchen
 */
public class HEFTPartitioner extends PurePartitioner{

    private Map EST;
    //private Map weight;
    private Map EFT;
    //private Map ECT;
    //private Map FAT;
    private Map AFT;
    private Map avail;
    private Map availMin;
    public HEFTPartitioner(ADag mDag, PegasusBag mBag)
    {
        super(mDag, mBag);
        //ECT = new HashMap<String, Double>();
        //FAT = new HashMap<String, Double>();
        //weight = new HashMap<Double, Double>();
        AFT = new HashMap<DAXAbstraction, Double>();
        EST = new HashMap<DAXAbstraction, Map>();
        avail = new HashMap<SiteCatalogEntry, Map>();
        availMin = new HashMap<SiteCatalogEntry, Integer>();
    }

    private double getAFT(edu.isi.pegasus.planner.classes.Job dagJob)
    {


         if(AFT.containsKey(dagJob))
             return (Double)AFT.get(dagJob);
        double runtime1 = 1e8;//big enough
        SiteCatalogEntry slot = null;
        int selectedSite = -1;
        for(int j = 0; j < mSiteIndex.size(); j ++)
        {
            SiteCatalogEntry site = (SiteCatalogEntry)mSiteIndex.get(j);
            //double runtime = (Double)weight.get(rank);
            double runtime = getweight(dagJob, site);
            double runtime2  = runtime + getEST(dagJob , site);// + communication;
            double space= Double.parseDouble(site.getEnvironmentVariable("SITESIZE")) * 1e8;

            //System.out.println("Estimated Runtime:" + (runtime+runtime2) +  " for " + dagJob.jobID);
            if(runtime2 >= 1e8)
                System.err.println("Runtime of this job is too large\n");
            if(runtime2 < runtime1)
            {
                double requiredSpace = getSizeMap(dagJob, j) + getSizeMap(j);
                if(space >= requiredSpace *0.95)
                {
                    runtime1 = runtime2;
                    slot = site;
                    selectedSite = j;
                }
                else
                {
                    System.out.println("The size space is not enough");
                }
            }

        }

        setavail(dagJob, slot, runtime1);

        

        AFT.put(dagJob, runtime1);
        addJob2Site(dagJob, selectedSite);
        System.out.println("job " + dagJob.jobID +" is mapped to " + slot.getSiteHandle() + " runtime1 " + runtime1 );


        return runtime1;
    }

    protected void setavail(edu.isi.pegasus.planner.classes.Job dagJob, SiteCatalogEntry site, double runtime)
    {
        if(!avail.containsKey(site))
            avail.put(site, new HashMap<Integer, Double>());
        if(!availMin.containsKey(site))
            availMin.put(site, 0);
        Map slotMap = (HashMap)avail.get(site);
        int indexMin = (Integer)availMin.get(site);

        double minTime = (Double)slotMap.get(indexMin);
        double estTime = getEST(dagJob , site);
        if( minTime <= estTime )
        {
            //is bounded by parent jobs
            double maxAFT = 0.0;
            ArrayList list = (ArrayList)mParentsMap.get(dagJob.jobID);
            for(Iterator it = list.iterator(); it.hasNext();)
            {
                edu.isi.pegasus.planner.classes.Job parent =
                        (edu.isi.pegasus.planner.classes.Job)mIDorLogicalID2Job.get((String)it.next());
                //double prank = (Double)job2rank.get(parent);
                double aft = getAFT(parent);
                SiteCatalogEntry parentSite = (SiteCatalogEntry)mJob2Site.get(parent);

                
                if (parentSite.equals(site)&&maxAFT<aft)
                {
                    maxAFT = aft;
                }
            }
            int slotNum   = Integer.parseInt(site.getEnvironmentVariable("SLOT"));
            for(int i = 0;i < slotNum; i++)
            {
                double run_t = (Double)slotMap.get(i);
                if(run_t == maxAFT)
                {
                    slotMap.put(i, runtime);
                    break;
                }
            }

            
        }
        else
        {
            //is bounded by resource
            slotMap.put(indexMin, runtime);
        }


       // slotMap.put(indexMin, runtime);
        //don't need to set indexMin beacuse getavail would re-calculate it again.

    }
    protected double getavail(SiteCatalogEntry site)
    {
        if(!avail.containsKey(site))
            avail.put(site, new HashMap<Integer, Double>());
        if(!availMin.containsKey(site))
            availMin.put(site, 0);
        Map slotMap = (HashMap)avail.get(site);
        int indexMin = (Integer)availMin.get(site);
        double minAvail = 1e8;//big enough
        int slot    = Integer.parseInt(site.getEnvironmentVariable("SLOT"));
        for(int i = 0; i < slot; i ++)
        {
            if(!slotMap.containsKey(i))
                slotMap.put(i, 0.0);
            double slotAvail = (Double)slotMap.get(i);
            if(slotAvail < minAvail)
            {
                minAvail = slotAvail;
                indexMin = i;
            }
        }
        availMin.put(site, indexMin);
        return minAvail;
    }
    protected double getweight(edu.isi.pegasus.planner.classes.Job dagJob, SiteCatalogEntry site)
    {
        if(!mJob2Time.containsKey(dagJob.jobID))
            return 0.0;
        double runtime = (Double)mJob2Time.get(dagJob.jobID);
        return runtime;
    }
    private double getEST(edu.isi.pegasus.planner.classes.Job dagJob, SiteCatalogEntry site)
    {
        if(!EST.containsKey(dagJob))
            EST.put(dagJob, new HashMap<SiteCatalogEntry, Double>());
        Map map = (HashMap)EST.get(dagJob);

        double value = 0.0;
        if(!map.containsKey(site))
        {
            double maxAFT = getavail(site);//?????
            if(!mParentsMap.containsKey(dagJob.jobID))
                mParentsMap.put(dagJob.jobID, new ArrayList<String>());
            ArrayList list = (ArrayList)mParentsMap.get(dagJob.jobID);

            for(Iterator it = list.iterator(); it.hasNext();)
            {
//                edu.isi.pegasus.planner.classes.Job parent =
//                (edu.isi.pegasus.planner.classes.Job)it.next();

                edu.isi.pegasus.planner.classes.Job parent =
                        (edu.isi.pegasus.planner.classes.Job)mIDorLogicalID2Job.get((String)it.next());
                //double prank = (Double)job2rank.get(parent);
                double aft = getAFT(parent);
                SiteCatalogEntry parentSite = (SiteCatalogEntry)mJob2Site.get(parent);

                double cmi = 0.0;
                if (parentSite.equals(site))
                    cmi = 0.0;
                else
                {
                    cmi = getSizeMap(dagJob) / 10e6;
                    //cmi = 1e4;
                }
                aft += cmi;
                double tmp_f = getavail(site);
                if(tmp_f > maxAFT)maxAFT = tmp_f;

                if(aft > maxAFT) maxAFT = aft;
            }


            value = maxAFT;
            map.put(site, value);
        }
        else
            value = (Double)map.get(site);
        return value;
    }



    @Override
    public void run()
    {
        super.run();//nust be done
        edu.isi.pegasus.planner.classes.Job rearNode = getRear();
        getAFT(rearNode);
       
        System.out.println("Jobs are all partitioned");

        
    }

}
